package cn.sunchis.demo.hadoop;

import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.List;

import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.util.Bytes;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.ImportResource;
import org.springframework.context.annotation.Lazy;
import org.springframework.data.hadoop.hbase.HbaseTemplate;
import org.springframework.data.hadoop.hbase.TableCallback;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

/**
 * Spring Hadoop DEMO
 * @Author 许亮
 * @Create 2018-7-5 15:44:10
 * {@link https://spring.io/projects/spring-hadoop#overview}
 * {@link http://hbase.apache.org/book.html#basic.prerequisites}
 */
@SuppressWarnings("deprecation")
@RestController
@SpringBootApplication
@ImportResource(locations = {"classpath:hbase-spring.xml"})
public class HadoopDemoApplication {
	@Autowired
	@Lazy(true)
	private HbaseTemplate hbaseTemplate;
	
	private int it;
	
	public static void main(String[] args) {
		System.setProperty("hadoop.home.dir", "E:\\dev-runtimes\\hadoop-common-2.2.0");

		SpringApplication.run(HadoopDemoApplication.class, args);
	}
	
	@RequestMapping(value="/")
	public String home() {
		// 先创建 user 表
		// create 'user',  {NAME=>'info'}
		
		//hbaseTemplate.put("user", "1", "info", "name", Bytes.toBytes("Alice"));
		
		Charset charset = Charset.forName("UTF-8");
		long t1 = System.currentTimeMillis();
		
		for (int i = 0; i < 200; i++) {
			it = i;
			
			hbaseTemplate.execute("user", new TableCallback<Object>() {
				@Override
				public Object doInTable(HTableInterface table) throws Throwable {
					int index;
					List<Put> puts = new ArrayList<>();
					
					for (int j = 0; j < 100000; j++) {
						index = 1 + j + it * 100000;
						puts.add(new Put(String.valueOf(index).getBytes(charset)).add("info".getBytes(charset), 
								"name".getBytes(charset), 
								Bytes.toBytes("Alice" + index)));
					}
					
					table.put(puts);
					return null;
				}
			});
		}
		
		long t2 = System.currentTimeMillis();
		
//        List<String> rows = hbaseTemplate.find("user", "info", "name", new RowMapper<String>() { 
//            public String mapRow(Result result, int i) throws Exception { 
//                return result.toString(); 
//            } 
//        });
//		
//		if (rows == null || rows.size() == 0) {
//			return "spring-boot-hadoop-demo";
//		}
//		
//		return rows.get(0);
		
		String string = "耗时：" + (t2 - t1)/1000.000 + "秒";
		return string;
	}
}
